package com;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class MyObservable {
    static public void main(String[] argv) {

        //                Future<Integer> future = threadPool.submit(new Callable<Integer>() {
//                    public Integer call() throws Exception {
//
//                        return new Random().nextInt(100);
//                    }
//                });








//
        ExecutorService threadPool1 = Executors.newSingleThreadExecutor();
        ExecutorService threadPool2 = Executors.newSingleThreadExecutor();
//
//
//
//
//
//
//        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
//
//            @Override
//            public void subscribe(final ObservableEmitter<Integer> emitter) throws Exception {
//                Thread.sleep(2000);
//                emitter.onNext(0);
//
//                emitter.onComplete();
//            }
//        });
//
//
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(Integer s) {
                System.out.println("onNext:" + s);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };
//
//        observable
//                .subscribeOn(Schedulers.from(threadPool))
//                .observeOn(Schedulers.newThread())
//                .subscribe(observer);


//        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                System.out.println("kaelpu" + "Observable thread is : " + Thread.currentThread().getName());
//                System.out.println("kaelpu" + "emitter 1");
//                emitter.onNext(1);
//            }
//        });
//
//        Consumer<Integer> consumer = new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Exception {
//                System.out.println("kaelpu" + "Observer thread is :" + Thread.currentThread().getName());
//                System.out.println("kaelpu" + "onNext: " + integer);
//            }
//        };
//
//        observable
//                .subscribeOn(Schedulers.from(threadPool))
//                .observeOn(Schedulers.newThread())
//                .subscribe(consumer);


        final ApiObservable apiObservable = new ApiObservable();

        Observable<Integer> observable = apiObservable.startPrint();

        observable = observable.subscribeOn(Schedulers.from(threadPool1));
        if (true) {
            observable = observable.doOnNext(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println("onNext1111:" + integer);
                }
            });
        }

        observable.concatMap(new Function<Integer, ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> apply(Integer integer) throws Exception {
                return apiObservable.stopPrint();
            }
        }).subscribe(observer);

//
//        apiObservable
//                .startPrint()
//                .subscribeOn(Schedulers.from(threadPool1))
//
//                .observeOn(Schedulers.from(threadPool1))
//                .concatMap(new Function<Integer, ObservableSource<Integer>>() {
//                    @Override
//                    public ObservableSource<Integer> apply(Integer integer) throws Exception {
//                        return apiObservable.stopPrint();
//                    }
//                })
//                .subscribe(observer);


        System.out.println("finish");
    }


}



